package com.wuwangfu.window;

import org.apache.flink.streaming.api.datastream.AllWindowedStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;

/**
 * @Author jcshen
 * @Date 2023-02-23
 * @PackageName:com.wuwangfu.window
 * @ClassName: CountWindowAllApply
 * @Description:
 * @Version 1.0.0
 *
 * https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/dev/datastream/operators/overview/#window-apply
 * 不分组，划分窗口
 * 然后调用apply对窗口内的数据进行处理
 * 就是将窗口内的数据先存起来（window state）中，当满足触发条件后，再将状态中的数据取出来进行计算
 */
public class CountWindowAllApply {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> line = env.socketTextStream("localhost", 8888);

        SingleOutputStreamOperator<Integer> maped = line.map(Integer::parseInt);
        //划分窗口
        AllWindowedStream<Integer, GlobalWindow> windowed = maped.countWindowAll(5);

        windowed.apply(new AllWindowFunction<Integer, Integer, GlobalWindow>() {
            /**
             *
             * @param window
             * @param values 输入的数据
             * @param out 输出的数
             * @throws Exception
             *
             * apply方法，将数据先存起来，满足窗口后再进行处理
             */
            @Override
            public void apply(GlobalWindow window, Iterable<Integer> values, Collector<Integer> out) throws Exception {
                /*全量聚合，满足窗口后，才全部取出再聚合，效率低*/
//                Integer sum = 0;
//                for (Integer value : values){
//                    sum += value;
//                }
//                out.collect(sum);
                /*全量聚合，满足窗口后，才全部取出再聚合，效率低*/
/*-------------------------------------------------------------------*/
                /*窗口内的数据，比较大小后再输出*/
                List<Integer> list = new ArrayList<>();
                for (Integer value : values){
                    list.add(value);
                }
                list.sort(new Comparator<Integer>() {
                    @Override
                    public int compare(Integer o1, Integer o2) {
                        //按照升序排
                        return Integer.compare(o1,o2);
                    }
                });
                for (Integer value : list){
                    out.collect(value);
                }
                /*窗口内的数据，比较大小后再输出*/
            }

        }).print()
        //不设置并行度的话，控制台打印的结果是不会排序的，因为并行度不一致导致的
        .setParallelism(1);

        env.execute();
    }
}
